package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.hmdp.constant.GlobalConstant;
import com.hmdp.entity.dto.Result;
import com.hmdp.entity.dto.UserDTO;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.utils.MyUtil;
import com.hmdp.utils.RedisIdWork;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Service
@Slf4j
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    ISeckillVoucherService iSeckillVoucherService;
    @Autowired
    MyUtil myUtil;
    @Autowired
    RedisIdWork redisIdWork;
    @Resource
    RedissonClient redissonClient;
    private IVoucherOrderService proxy=null;
    private BlockingQueue<VoucherOrder> blockingQueue=new ArrayBlockingQueue<>(1024*1024);
    private static final ExecutorService SECKILL_ORDER_EXECUTOR= Executors.newSingleThreadExecutor();
    @PostConstruct
    private void init(){
        Long voucherId=1L;
        LambdaQueryWrapper<SeckillVoucher> lqw=new LambdaQueryWrapper<>();
        lqw.eq(SeckillVoucher::getVoucherId,voucherId);
        SeckillVoucher seckVoucher = iSeckillVoucherService.getOne(lqw);
        redisTemplate.opsForValue().set(GlobalConstant.SECKILL_STOCK+voucherId,seckVoucher.getStock().toString());
//        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandler());
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderHandlerRedisStream());
    }
    private class VoucherOrderHandler implements Runnable{

        @Override
        public void run() {
           while (true)
           {
               try {
                   VoucherOrder voucherOrder = blockingQueue.take();
                   log.info("订单处理中...");
                   handleVoucherOrder(voucherOrder);
               } catch (InterruptedException e) {
                   log.error("订单处理异常",e);
               }
           }
        }
    }
    private class VoucherOrderHandlerRedisStream implements Runnable{

        @Override
        public void run() {
            while (true)
            {
                try{
                    List<MapRecord<String,Object,Object>> list = redisTemplate.opsForStream().read(
                            Consumer.from(GlobalConstant.GROUP_PREFIX + "1", GlobalConstant.CONSUMER_PREFIX + "1"),
                            StreamReadOptions.empty().count(1),
                            StreamOffset.create(GlobalConstant.STREAM_NAME, ReadOffset.lastConsumed()));
                    if(list==null||list.isEmpty())
                    {
                        continue;
                    }
                    MapRecord<String,Object,Object> record=list.get(0);
                    Map<Object,Object> map=record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(map, new VoucherOrder(), true);
                    handleVoucherOrder(voucherOrder);
                    redisTemplate.opsForStream().acknowledge(GlobalConstant.STREAM_NAME,GlobalConstant.GROUP_PREFIX+"1",record.getId());
                }catch (Exception e){
                    log.error("订单处理异常",e);
                    handlePedingList();
                }

            }
        }
    }
    public void handlePedingList()
    {
        while (true)
        {
            try{
                List<MapRecord<String,Object,Object>> list = redisTemplate.opsForStream().read(
                        Consumer.from(GlobalConstant.GROUP_PREFIX + "1", GlobalConstant.CONSUMER_PREFIX + "1"),
                        StreamReadOptions.empty().count(1),
                        StreamOffset.create(GlobalConstant.STREAM_NAME, ReadOffset.from("0")));
                if(list==null||list.isEmpty())
                {
                    break;
                }
                MapRecord<String,Object,Object> record=list.get(0);
                Map<Object,Object> map=record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(map, new VoucherOrder(), true);
                handleVoucherOrder(voucherOrder);
                redisTemplate.opsForStream().acknowledge(GlobalConstant.STREAM_NAME,GlobalConstant.GROUP_PREFIX+"1",record.getId());
            }catch (Exception e){
                log.error("peding_list处理异常",e);
                try {
                    Thread.sleep(20);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }

        }
    }

    private void handleVoucherOrder(VoucherOrder voucherOrder)
    {
        Long userId = voucherOrder.getUserId();
        RLock lock=redissonClient.getLock(GlobalConstant.LOCK_ORDER_PREFIX+userId);
        if(!lock.tryLock())
        {
            log.error(GlobalConstant.DO_NOT_REPEAT);
            return;
        }
        proxy.saveVoucherOrderAsync(voucherOrder);
        lock.unlock();
    }
    static private final DefaultRedisScript<Long> redisScript;
    static
    {
        redisScript=new DefaultRedisScript<>();
        redisScript.setLocation(new ClassPathResource(GlobalConstant.JUDGE_SCRIPT));
        redisScript.setResultType(Long.class);
    }

    @Override
    public Result seckillVoucher(Long voucherId) {
        Long userId= UserHolder.getUser().getId();
        if(voucherId==null)
        {
            return Result.fail(GlobalConstant.PARAMS_EXCEPTION);
        }
        LambdaQueryWrapper<SeckillVoucher> lqw = new LambdaQueryWrapper<>();
        lqw.eq(SeckillVoucher::getVoucherId, voucherId);
        SeckillVoucher seckillVoucher = iSeckillVoucherService.getOne(lqw);
        if(seckillVoucher==null)
        {
            return Result.fail(GlobalConstant.PARAMS_EXCEPTION);
        }
        if(seckillVoucher.getStock()<=0)
        {
            return Result.fail(GlobalConstant.VOUCHER_STOCK_NULL);
        }
        if(!LocalDateTime.now().isAfter(seckillVoucher.getBeginTime()))
            return Result.fail(GlobalConstant.TIME_BEFORE);
        if(!LocalDateTime.now().isBefore(seckillVoucher.getEndTime()))
            return Result.fail(GlobalConstant.TIME_AFTER);
        RLock myLock=redissonClient.getLock(GlobalConstant.SHOP_ID_PREFIX + userId);
        if(!myLock.tryLock())
        {
            return Result.fail(GlobalConstant.TRY_AGAIN_LATER);
        }
        try{
            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
            return proxy.createVoucher(voucherId);
        }finally {
            myLock.unlock();
        }
//        MyLock myLock = new MyLock(GlobalConstant.SHOP_ID_PREFIX + userId, redisTemplate);
//        if(!myLock.tryLock(GlobalConstant.KEY_TTL))
//        {
//            return Result.fail(GlobalConstant.TRY_AGAIN_LATER);
//        }
//        try{
//            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
//            return proxy.createVoucher(voucherId);
//        }finally {
//            myLock.unlock();
//        }
//        synchronized (userId.toString().intern())
//        {
//            IVoucherOrderService proxy = (IVoucherOrderService)AopContext.currentProxy();
//            return proxy.createVoucher(voucherId);
//        }
    }
    @Transactional
    public Result createVoucher(Long voucherId)
    {
        Long userId=UserHolder.getUser().getId();
        LambdaQueryWrapper<VoucherOrder> lqw2=new LambdaQueryWrapper<>();
        lqw2.eq(VoucherOrder::getVoucherId,voucherId)
                .eq(VoucherOrder::getUserId,userId);
        List<VoucherOrder> list = this.list(lqw2);
        if(!list.isEmpty())
        {
            return Result.fail(GlobalConstant.DO_NOT_REPEAT);
        }
        boolean update = iSeckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id",voucherId).gt("stock",0).update();
        if(!update)
            return Result.fail(GlobalConstant.FAIL);
        VoucherOrder voucherOrder = new VoucherOrder();
        LocalDateTime now = LocalDateTime.now();
        voucherOrder.setId(redisIdWork.getId());
        voucherOrder.setUserId(userId);
        voucherOrder.setVoucherId(voucherId);
        voucherOrder.setPayType(3);
        voucherOrder.setStatus(2);
        voucherOrder.setCreateTime(now);
        voucherOrder.setUpdateTime(now);
        this.save(voucherOrder);
        return Result.ok();
    }
    @Override
    public Result seckillVoucherAsync(Long voucherId)
    {
        UserDTO user = UserHolder.getUser();
        Long result = (Long)redisTemplate.execute(redisScript, Collections.emptyList(), voucherId.toString(), user.getId().toString());
        int r=result.intValue();
        if(r!=0)
        {
            return Result.fail(r==1?GlobalConstant.VOUCHER_STOCK_NULL:GlobalConstant.DO_NOT_REPEAT);
        }

        Long id = redisIdWork.getId();
        VoucherOrder voucherOrder = new VoucherOrder();
        voucherOrder.setId(id);
        voucherOrder.setUserId(user.getId());
        voucherOrder.setVoucherId(voucherId);
        if(proxy==null)
            proxy = (IVoucherOrderService)AopContext.currentProxy();
        blockingQueue.add(voucherOrder);
        log.info("订单："+id);
        return Result.ok(id);
    }
    @Override
    @Transactional
    public void saveVoucherOrderAsync(VoucherOrder voucherOrder)
    {
        Long userId=voucherOrder.getUserId();
        Long voucherId=voucherOrder.getVoucherId();
        LambdaQueryWrapper<VoucherOrder> lqw2=new LambdaQueryWrapper<>();
        lqw2.eq(VoucherOrder::getVoucherId,voucherId)
                .eq(VoucherOrder::getUserId,userId);
        List<VoucherOrder> list = this.list(lqw2);
        if(!list.isEmpty())
        {
            log.error(GlobalConstant.DO_NOT_REPEAT);
        }
        boolean update = iSeckillVoucherService.update().setSql("stock = stock - 1").eq("voucher_id",voucherId).gt("stock",0).update();
        if(!update)
           log.error(GlobalConstant.FAIL);
        LocalDateTime now = LocalDateTime.now();
        voucherOrder.setPayType(3);
        voucherOrder.setStatus(2);
        voucherOrder.setCreateTime(now);
        voucherOrder.setUpdateTime(now);
        this.save(voucherOrder);
        log.info(GlobalConstant.ORDER_BUILD_SUCCESS);
    }
    @Override
    public Result seckillVoucherAsyncRedisStream(Long voucherId)
    {
        UserDTO user = UserHolder.getUser();
        Long id = redisIdWork.getId();
        Long result = (Long)redisTemplate.execute(redisScript, Collections.emptyList(), voucherId.toString(), user.getId().toString(),id.toString());
        int r=result.intValue();
        if(r!=0)
        {
            return Result.fail(r==1?GlobalConstant.VOUCHER_STOCK_NULL:GlobalConstant.DO_NOT_REPEAT);
        }
        if(proxy==null)
            proxy = (IVoucherOrderService)AopContext.currentProxy();
        log.info("订单："+id);
        return Result.ok(id);
    }

}
